/**
 * Licensed under:
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * @author Emanuele Gherardini (e.gherardini@yahoo.it / e.gherardini@hotmail.com)
 */
package org.apache.camel.component.coherence.queue;

import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.oracle.coherence.common.identifiers.Identifier;
import com.oracle.coherence.patterns.messaging.DefaultMessagingSession;
import com.oracle.coherence.patterns.messaging.MessagingSession;
import com.oracle.coherence.patterns.messaging.QueueSubscriptionConfiguration;
import com.oracle.coherence.patterns.messaging.SubscriptionConfiguration;

public class CoherenceQueueEndpoint extends DefaultEndpoint {

	private final String queueName;
	private final Identifier queueIdentifier;
	private final MessagingSession messagingSession;
	private final QueueSubscriptionConfiguration subscriptionConfiguration;
	private final int concurrentConsumers;
	
	private static final Logger log = LoggerFactory.getLogger(CoherenceQueueEndpoint.class);
	
	/**
	 * Creates a new CoherenceQueueEndpoint getting a new {@link DefaultmessagingSession} instance
	 * and without durable subscribers.
	 * 
	 * @param uri
	 * @param comp
	 * @param queueName
	 * 			the name of the queue
	 */
	public CoherenceQueueEndpoint(String uri, Component comp, String queueName) {
		this(uri,comp,queueName, DefaultMessagingSession.getInstance(), false, 1);
	}
	
	/**
	 * Creates a new CoherenceQueueEndpoint getting a new {@link DefaultmessagingSession} instance
	 * and configuring durable subscriptions based on the given parameter.
	 * @param uri
	 * @param comp
	 * @param queueName
	 * @param durableSubscriber
	 */
	public CoherenceQueueEndpoint(String uri, Component comp, String queueName, boolean durableSubscriber) {
		this(uri,comp,queueName, DefaultMessagingSession.getInstance(), durableSubscriber, 1);
	}
	
	/**
	 * Creates a new CoherenceQueueEndpoint getting a new {@link DefaultmessagingSession} instance
	 * and configuring durable subscriptions based on the given parameter.
	 * @param uri
	 * @param comp
	 * @param queueName
	 * @param durableSubscriber
	 * @param concurrentConsumers
	 */
	public CoherenceQueueEndpoint(String uri, Component comp, String queueName, boolean durableSubscriber, int concurrentConsumers) {
		this(uri,comp,queueName, DefaultMessagingSession.getInstance(), durableSubscriber, concurrentConsumers);
	}
	
	/**
	 * Creates a new CoherenceQueueEndpoint with the given {@link MessagingSession} instance
	 * and configuring durable subscriptions based on the given parameter.
	 * 
	 * @param uri
	 * @param comp
	 * @param queueName
	 * @param messagingSession
	 * @param durableSubscribers
	 */
	public CoherenceQueueEndpoint(String uri, Component comp, String queueName, MessagingSession messagingSession, boolean durableSubscribers, int concurrentConsumers) {
		super(uri, comp);
		this.queueName = queueName;
		this.messagingSession = messagingSession;
		this.concurrentConsumers = concurrentConsumers;
		
		if(durableSubscribers) {
			this.subscriptionConfiguration = QueueSubscriptionConfiguration.newDurableConfiguration(queueName+"_durableSubscription");
		} else {
			this.subscriptionConfiguration = QueueSubscriptionConfiguration.newNonDurableConfiguration();
		}
		
		this.queueIdentifier = messagingSession.createQueue(queueName);
		
		log.info("CoherenceQueueEndpoint succesfully created: " + this);
	}
	
	@Override
	public Producer createProducer() throws Exception {
		return new CoherenceQueueProducer(this);
	}
	
	@Override
	public Consumer createConsumer(Processor processor) throws Exception {
		return new CoherenceQueueConsumer(this, subscriptionConfiguration, processor, concurrentConsumers);
	}

	@Override
	public boolean isSingleton() {
		return true;
	}

	public String getQueueName() {
		return queueName;
	}

	public Identifier getQueueIdentifier() {
		return queueIdentifier;
	}

	public MessagingSession getMessagingSession() {
		return messagingSession;
	}

	public SubscriptionConfiguration getSubscriptionConfiguration() {
		return subscriptionConfiguration;
	}
	
	public int getConcurrentConsumers() {
		return concurrentConsumers;
	}

	@Override
	public String toString() {
		return "CoherenceQueueEndpoint [queueName=" + queueName
				+ ", subscriptionConfiguration=" + subscriptionConfiguration + "] concurrentConsumers = " + concurrentConsumers;
	}
	
	
}
